package org.lzfto.flink.demo.business.flinkTest.flinkObjects.trigger;

import java.time.Duration;
import java.util.Date;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.lzfto.flink.demo.utils.TimeUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 自定义全局窗口Stream触发器，在指定时间之后会清理窗口内容(在此期间会定时触发)
 * </br> 注意，这个是多线程下运行的，所以需要注意处理局部变量。大部分Flink的类都有这个问题。
 * @author lzfto
 * @创建时间 2025年6月20日 14:24:17
 */
public class  TimeLimitTrigger<T> extends Trigger<T, GlobalWindow> {
	private static final long serialVersionUID = 1L;

	private final Logger LOGGER = LoggerFactory.getLogger(getClass());
	private final long maxRunMillis; // 最大运行时长
	private final long repeatInterval; // 启动后每次运行的间隔（毫秒)
	// 定义状态描述符
	private final ValueStateDescriptor<Long> startTimeStateDesc = new ValueStateDescriptor<>("startTime", Long.class);
	private final ValueStateDescriptor<Long> lastFireTimeStateDesc = new ValueStateDescriptor<>("lastFireTime",
			Long.class);

	/**
	 * 构造函数
	 * 
	 * @param duration       启动后持续的时间
	 * @param repeatInterval 启动后每次运行的间隔（毫秒)
	 */
	public TimeLimitTrigger(Duration duration, Long repeatInterval) {
		LOGGER.warn("自定义触发器将持续运行{}秒", duration.getSeconds());
		this.maxRunMillis = duration.toMillis();
		this.repeatInterval = repeatInterval;
	}

	@Override
	public TriggerResult onElement(T element, long timestamp, GlobalWindow window, TriggerContext ctx)
			throws Exception {
		ValueState<Long> startTimeState = ctx.getPartitionedState(startTimeStateDesc);
		ValueState<Long> lastFireTimeState = ctx.getPartitionedState(lastFireTimeStateDesc);
		Long startTime = startTimeState.value();
		long now = ctx.getCurrentProcessingTime();

		if (startTime == null) {
			startTime = now;
			startTimeState.update(startTime);
			lastFireTimeState.update(now);

			ctx.registerProcessingTimeTimer(startTime + repeatInterval); // 首次周期触发
			ctx.registerProcessingTimeTimer(startTime + maxRunMillis); // 最终终止时间
			LOGGER.warn("自定义触发器开始运行于:{}", TimeUtil.formatDate(new Date(startTime)));
		}

		return TriggerResult.CONTINUE;
	}

	@Override
	public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
		ValueState<Long> startTimeState = ctx.getPartitionedState(startTimeStateDesc);
		ValueState<Long> lastFireTimeState = ctx.getPartitionedState(lastFireTimeStateDesc);

		Long startTime = startTimeState.value();
		Long lastFireTime = lastFireTimeState.value();

		if (startTime == null || lastFireTime == null) {
			return TriggerResult.CONTINUE;
		}

		if (time >= startTime + maxRunMillis) {
			// 超出最大运行时长：触发并清空窗口
			clear(window, ctx);
			LOGGER.warn("自定义触发器已经持续运行{}秒，将要关闭", maxRunMillis/1000);
			return TriggerResult.FIRE_AND_PURGE;
		} else if (time >= lastFireTime + repeatInterval) {
			// 正常周期性触发
			lastFireTimeState.update(time);
			ctx.registerProcessingTimeTimer(time + repeatInterval); // 注册下一次
			return TriggerResult.FIRE;
		}

		return TriggerResult.CONTINUE;

	}

	@Override
	public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) {
		//由于没有注册时间时间，所以这个就这样吧！
		return TriggerResult.CONTINUE;
	}

	@Override
	public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {
		// 可选清理操作
		ctx.getPartitionedState(startTimeStateDesc).clear();
		ctx.getPartitionedState(lastFireTimeStateDesc).clear();
		LOGGER.warn("触发器环境已经清除自定义状态:开始时间、最后运行时间");
	}
}
